package com.lagou.client;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author yangpeng
 * @time 2020/5/21
 * @since 1.0.0
 */
public class ZkProxy {

    public static final Map<String,ServerInfo> SERVER_INFO = new HashMap<>();

    private static CuratorFramework client;

    static {
        //创建重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
        //启动与zk的连接
        client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(retryPolicy)
                .namespace("cluster")
                .build();
        client.start();
    }

    public static CuratorFramework getClient() {
        return client;
    }

    public void loadZkInfo(){
        try {
            CuratorFramework client = getClient();
            //获取所有子节点信息
            List<String> childrenPath = client.getChildren().forPath("/");
            //开始创建连接存储Map
            for (String childPath : childrenPath) {
                String[] split = childPath.split(":");
                SERVER_INFO.put(childPath,new ServerInfo(split[0],Integer.valueOf(split[1])));
            }
            //添加监听
            watchChildrenNode(client);
        }catch(Exception ex){
            ex.printStackTrace();
        }
    }

    /**
     * 监听服务端节点的状态变化
     * @param client
     * @throws Exception
     */
    private void watchChildrenNode(CuratorFramework client) throws Exception {
        PathChildrenCache cache = new PathChildrenCache(client, "/", true);
        cache.start();
        System.out.println("注册根节点监听事件~~~~~~");
        PathChildrenCacheListener listener = (curatorFramework, event) -> {
            String path = event.getData().getPath();
            String[] split = path.split(":");
            String ip = split[0].substring(1);
            Integer port = Integer.valueOf(split[1]);
            switch (event.getType()) {
                case CHILD_ADDED:
                    //增加连接
                    System.out.println("Host: " + ip + " port: " + port + " 上线了...");
                    SERVER_INFO.put(path.substring(1),new ServerInfo(ip,port));
                    break;
                case CHILD_REMOVED:
                    //断开连接
                    SERVER_INFO.put(path.substring(1),new ServerInfo(ip,port));
                    System.out.println("Host: " + ip + " port: " + port + " 离线了...");
                    break;
                default:
                    break;
            }
        };
        cache.getListenable().addListener(listener);
    }
}
